#include "swoole.h"
#include "Server.h"

static int swReactorThread_onRead(swReactor *reactor, swEvent *ev);
static int swReactorThread_onWrite(swReactor *reactor, swEvent *ev);
static int swReactorThread_onPackage(swReactor *reactor, swEvent *event);

static int swReactorThread_loop(swThreadParam *param);

int swReactorThread_onClose(swReactor *reactor, swEvent *event){
	swServer *serv = reactor->ptr;

	int fd = event->fd;
	swDataHead notify_ev;
	bzero(&notify_ev, sizeof(notify_ev));

	notify_ev.from_id = reactor->id;
	notify_ev.fd = fd;
	notify_ev.type = SW_EVENT_CLOSE;

	if(reactor->del(reactor,fd) == 0){
		return SwooleG.factory->notify(SwooleG.factory, &notify_ev);
	}else{
		return -1;
	}
}

int swReactorThread_send(swSendData *_send){
	swServer *serv = SwooleG.serv;
    uint32_t session_id = _send->info.fd;
    void *_send_data = _send->data;
    uint32_t _send_length = _send->length;
	int n;

    swConnection *conn;
	conn = swServer_connection_verify(serv, session_id);

	int fd = conn->fd;
	swReactor *reactor;

	reactor = &(serv->reactor_threads[conn->from_id].reactor);
	assert(fd % serv->reactor_num == reactor->id);

	if(_send->info.type == SW_EVENT_CLOSE){
		printf("-=-=-=closing-=-=-=,%d\n",fd);
		//reactor->close(reactor,fd);
		swReactorThread_close(reactor,fd);
		printf("-=-=-=close-=-=-=\n");
		return 0;
	}

	n = swConnection_send(conn,_send_data,_send_length,0);
	if(n == _send_length){
		return 0;
	}

	return -1;
}

static int swReactorThread_onPipeReceive(swReactor *reactor, swEvent *ev){
	int n;
	swEventData resp;
	swSendData _send;

	n = read(ev->fd, &resp, sizeof(resp));
	memcpy(&_send.info, &resp.info, sizeof(resp.info));
	_send.data = resp.data;
	_send.length = resp.info.len;
	swReactorThread_send(&_send);
}

static int swReactorThread_onPipeWrite(swReactor *reactor, swEvent *ev){
}

int swReactorThread_create(swServer *serv){
	int ret;

	/**
     * init reactor thread pool
     */
    serv->reactor_threads = SwooleG.memory_pool->alloc(SwooleG.memory_pool, (serv->reactor_num * sizeof(swReactorThread)));
    if (serv->reactor_threads == NULL)
    {
        return -1;
    }

	serv->connection_list = sw_shm_calloc(serv->max_connection,sizeof(swConnection));

	ret = swFactoryProcess_create(&(serv->factory), serv->worker_num);
	return ret;
}

int swReactorThread_start(swServer *serv, swReactor *main_reactor_ptr){
	swListenPort *ls;
	ls = serv->listen_list;
	swThreadParam *param;
	swReactorThread *thread;
	pthread_t pidt;
	int i;

	swServer_store_listen_socket(serv);

	main_reactor_ptr->add(main_reactor_ptr,ls->sock,SW_FD_LISTEN);

	//pthread_barrier_init and why???

	for(i = 0; i < serv->reactor_num; i++){
		thread = &(serv->reactor_threads[i]);
		param = SwooleG.memory_pool->alloc(SwooleG.memory_pool, sizeof(swThreadParam));
        if (param == NULL)
        {
            return -1;
        }

        param->object = serv;
        param->pti = i;
		if(pthread_create(&pidt,NULL,(void * (*)(void *)) swReactorThread_loop,(void *)param) < 0){
			printf("pthread_create[tcp_reactor] failed. Error: %s[%d]",strerror(errno), errno);
		}

		thread->thread_id = pidt;
	}
}

static int swReactorThread_onRead(swReactor *reactor, swEvent *event){
	swConnection *conn =  event->socket;
	swDispatchData task;
	int n;
	swFactory *factory = SwooleG.factory;

	n = recv(conn->fd,task.data.data,SW_BUFFER_SIZE,0);
	if(n < 0){
		return 0;
	}else if(n == 0){
		//not write
		printf("fd closed\n");
		swReactorThread_onClose(reactor, event);
	}else{
		//do not think of big data...
		task.data.info.from_fd = conn->from_fd;
		task.data.info.fd = event->fd;
        task.data.info.from_id = event->from_id;
        task.data.info.len = n;
        task.data.info.type = SW_EVENT_PACKAGE_END;
        task.target_worker_id = -1;

		factory->dispatch(factory,&task);
	}

	return 0;
}

static int swReactorThread_onPackage(swReactor *reactor, swEvent *event){
}

static int swReactorThread_onWrite(swReactor *reactor, swEvent *ev){
	int fd = ev->fd;
	swServer *serv = SwooleG.serv;

	//get conn
	swConnection *conn = swServer_connection_get(serv,fd);
	if(conn == NULL || conn->active == 0){
		return -1;
	}

	if(conn->connect_notify){
		conn->connect_notify = 0;
		if(serv->onConnect){
			swServer_tcp_notify(serv,conn,SW_EVENT_CONNECT);
		}
	}

	return reactor->set(reactor,fd,SW_EVENT_TCP | SW_EVENT_READ);
}

void swReactorThread_set_protocol(swServer *serv, swReactor *reactor){
	//UDP Packet
    reactor->setHandle(reactor, SW_FD_UDP, swReactorThread_onPackage);
    //Write
    reactor->setHandle(reactor, SW_FD_TCP | SW_EVENT_WRITE, swReactorThread_onWrite);
    //Read
    reactor->setHandle(reactor, SW_FD_TCP | SW_EVENT_READ, swReactorThread_onRead);
}

static int swReactorThread_loop(swThreadParam *param){
	printf("swReactorThread_loop run\n");

	swServer *serv = SwooleG.serv;
	int reactor_id = param->pti;
	int ret;
	int i = 0, pipe_fd;

	pthread_t thread_id = pthread_self();

	swReactorThread *thread = swServer_get_thread(serv, reactor_id);
	swReactor *reactor = &thread->reactor;

	ret = swReactor_create(reactor, SW_REACTOR_MAXEVENTS);
	if(ret < 0){
		return -1;
	}

	reactor->ptr = serv;
	reactor->id = reactor_id;
	reactor->socket_list = serv->connection_list;
	reactor->max_socket = serv->max_connection;

	reactor->setHandle(reactor, SW_FD_CLOSE, swReactorThread_onClose);
	reactor->setHandle(reactor, SW_FD_PIPE | SW_EVENT_READ, swReactorThread_onPipeReceive);
	reactor->setHandle(reactor, SW_FD_PIPE | SW_EVENT_WRITE, swReactorThread_onPipeWrite);

	swReactorThread_set_protocol(serv, reactor);

	for(i = 0; i < serv->worker_num; i++){
		if(i % serv->reactor_num == reactor_id){
			//think of buffer
			pipe_fd = serv->workers[i].pipe_master;

			serv->connection_list[pipe_fd].from_id = reactor_id;
			serv->connection_list[pipe_fd].fd = pipe_fd;
		}

		swSetNonBlock(pipe_fd);
		reactor->add(reactor, pipe_fd, SW_FD_PIPE);
	}

	reactor->wait(reactor,NULL);
	reactor->free(reactor);
	pthread_exit(0);
}

int swReactorThread_send2worker(void *data, int len, uint16_t target_worker_id){
	swServer *serv = SwooleG.serv;

	assert(target_worker_id < serv->worker_num);

	int ret = -1;
	swWorker *worker = &(serv->workers[target_worker_id]);

	int pipe_fd = worker->pipe_master;
	int thread_id = serv->connection_list[pipe_fd].from_id;

	ret = write(pipe_fd,(void*) data,len);

	return ret;
}

int swReactorThread_close(swReactor *reactor, int fd){
	//do not think of buffer
	swConnection *socket = swReactor_get(reactor, fd);
	swServer *serv = reactor->ptr;

	bzero(socket,sizeof(swConnection));
	socket->removed = 1;

	swSession *session = swServer_get_session(serv, socket->session_id);
	session->fd = 0;
	bzero(socket,sizeof(swConnection));
	socket->removed = 1;

	return close(fd);
}

void swReactorThread_free(swServer *serv){
	int i;
	swReactorThread *thread;

	for(i = 0; i < serv->reactor_num; i++){
		thread = &(serv->reactor_threads[i]);
		pthread_cancel(thread->thread_id);
		if(pthread_join(thread->thread_id,NULL) < 0){
			printf("pthread join error\n");
		}
	}
}
